package cn.ilikes.tools.nosql.jpa.clients.indexs.index.hadoops;

import java.io.IOException;




import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.lucene.store.BufferedIndexInput;

public class FileSystemIndexInput extends BufferedIndexInput {

	// shared by clones
	private class Descriptor {
		public final FSDataInputStream in;
		public long position; // cache of in.getPos()

		public Descriptor(FileSystem fileSystem,Path file, int ioFileBufferSize) throws IOException {
			this.in = fileSystem.open(file);
		}
		
		public void close()
		{
//			try {
//				in.close();
//			} catch (IOException e) {
//				
//				e.printStackTrace();
//			}
		}
	}

	private final Path filePath; // for debugging
	private final Descriptor descriptor;
	private final long length;
	
	public FileSystemIndexInput(FileSystem fileSystem, Path path, String resourceDesc, int bufferSize) throws IOException {
		super(resourceDesc, bufferSize);
		filePath = path;
		descriptor = new Descriptor(fileSystem,path, bufferSize);
		length = fileSystem.getFileStatus(path).getLen();
		
	}

	@Override
	protected void readInternal(byte[] b, int offset, int length) throws IOException {
	          ;
		synchronized (descriptor) {
			long position = getFilePointer();
			if (position != descriptor.position) {
				descriptor.in.seek(position);
				descriptor.position = position;
			}
			int total = 0;
			do {
				int i = descriptor.in.read(b, offset + total, length - total);
				if (i == -1) {
					throw new IOException("Read past EOF");
				}
				descriptor.position += i;
				total += i;
			} while (total < length);
		}

	}

	

	@Override
	public void close() throws IOException {
		descriptor.close();

	}

	@Override
	public long length() {
		return length;
	}

	@Override
	protected void seekInternal(long pos) throws IOException {
		descriptor.position=descriptor.position+pos;
		
		
	}

	
}
